/*
 * @Author: wuqingchun
 * @Date: 2024-07-01 13:48:28
 * @Description: 
 * @LastEditors: wuqingchun
 * @LastEditTime: 2024-07-08 20:34:41
 */
#ifndef __ARQ_RECEIVER_H__
#define __ARQ_RECEIVER_H__

#include <thread>
#include <list>

#include "arq.h"
#include "arq_allocator.h"
#include "arq_socket.h"


// 接收窗口类模板
template <size_t BUFFER_SIZE = 1024>
class CARQReceiveWindow
{
    static constexpr size_t BUFFER_INDEX_MASK = BUFFER_SIZE - 1;
    static_assert((BUFFER_SIZE & BUFFER_INDEX_MASK) == 0, "BUFFER_SIZE is not power of 2");
public:
    CARQReceiveWindow()
    {

    }

    ~CARQReceiveWindow()
    {
        if (m_ppstUnAckMsg != nullptr)
        {
            delete [] m_ppstUnAckMsg;
            m_ppstUnAckMsg = nullptr;
        }
    }

    int32_t Init(uint32_t uiWndSize)
    {
        if (uiWndSize > BUFFER_SIZE)
        {
            // 设置窗口太大
            return -1;
        }

        m_ppstUnAckMsg = new (std::nothrow) ARQ_HEAD_S*[BUFFER_SIZE];
        if (m_ppstUnAckMsg == nullptr)
        {
            return -1;
        }

        m_uiWndSize = uiWndSize;

        return 0;
    }
    
    bool OnRecv(ARQ_HEAD_S* pstMsg)
    {
        //printf("Received message, seqno=%u, nextseqno=%u\n", pstMsg->uiSeqNo, m_uiNextSeqNo);
        // printf("Repeat: %lu, unorder: %lu, outwnd: %lu\n", m_ui64RepeatCount, m_ui64UnorderCount,
        //        m_ui64OutWndCount);
        
        bool bRet = IsValidMsg(pstMsg->uiSeqNo);
        if (!bRet)
        {
            return bRet;
        }
        
        if (pstMsg->uiSeqNo == m_uiNextSeqNo)
        {
            m_uiNextSeqNo++;
            return bRet;
        }

        m_ui64UnorderCount++;

        if (pstMsg->uiSeqNo > m_uiMaxSeqNo)
        {
            // 更新最大消息号
            m_uiMaxSeqNo = pstMsg->uiSeqNo;
        }

        bRet = false;
        m_ppstUnAckMsg[pstMsg->uiSeqNo & BUFFER_INDEX_MASK] = pstMsg;

        for (uint32_t uiSeqNo = m_uiNextSeqNo + 1; uiSeqNo <= m_uiMaxSeqNo; uiSeqNo++)
        {
            if (m_ppstUnAckMsg[uiSeqNo & BUFFER_INDEX_MASK] == nullptr)
            {
                break;
            }

            m_ppstUnAckMsg[uiSeqNo & BUFFER_INDEX_MASK] = nullptr;
            m_uiNextSeqNo++;

            bRet = true;
        }

        if (m_uiNextSeqNo == m_uiMaxSeqNo + 1)
        {
            m_ui64UnorderCount = 0;
        }

        return bRet;
    }

    uint32_t GetNextSeqNo()
    {
        return m_uiNextSeqNo;
    }

    bool IsValidMsg(uint32_t uiSeqNo)
    {
        if (uiSeqNo < m_uiNextSeqNo)
        {
            // 重复消息，丢弃
            m_ui64RepeatCount++;
            return false;
        }

        if (m_uiNextSeqNo + m_uiWndSize < uiSeqNo)
        {
            // 超出窗口大小，丢弃
            m_ui64OutWndCount++;
            return false;
        }

        return true;
    }
    
    void GetStatistics()
    {
        printf("WndSize  : %u\n", m_uiWndSize);
        printf("NextSeqNo: %u\n", m_uiNextSeqNo);
        printf("Unorder  : %lu\n", m_ui64UnorderCount);
        printf("Repeat   : %lu\n", m_ui64RepeatCount);
        printf("OutWindow: %lu\n", m_ui64OutWndCount);
        printf("\n");
    }
    
private:
    ARQ_HEAD_S** m_ppstUnAckMsg = nullptr;
    uint32_t m_uiWndSize = 0;
    uint32_t m_uiNextSeqNo = 0;
    uint32_t m_uiMaxSeqNo = 0;
    uint64_t m_ui64UnorderCount = 0;
    uint64_t m_ui64RepeatCount = 0;
    uint64_t m_ui64OutWndCount = 0;
};


class CARQReceiver : public CARQSocket
{
    static constexpr size_t ARQ_RECV_QUEUE_SIZE = 1024;
public:
    CARQReceiver();
    
    int32_t Init(SOCKADDR_UN& refPeerAddr, uint32_t uiWndSize);
    int32_t Start();

    void GetStatistics();

private:
    void __ThreadFuncRecv();

    int32_t __HandleMsg(ARQ_HEAD_S& refHead);

    int32_t __SendAck(uint32_t uiAckNo);
private:
    std::thread m_objRecvThread;
    volatile bool m_bRecvRuning = false;
    CARQAllocator m_objAllocator;    
    CARQReceiveWindow<> m_objRcvWnd;
};

#endif //__ARQ_RECEIVER_H__